package br.pucrs.lognplay.job;

import static org.elasticsearch.index.query.FilterBuilders.rangeFilter;
import static org.elasticsearch.index.query.QueryBuilders.filteredQuery;
import static org.elasticsearch.index.query.QueryBuilders.termQuery;

import java.util.Calendar;
import java.util.List;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.common.joda.time.DateTime;
import org.elasticsearch.common.joda.time.format.DateTimeFormat;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import br.pucrs.lognplay.entity.Client;
import br.pucrs.lognplay.entity.DynamicView;
import br.pucrs.lognplay.entity.LogSource;
import br.pucrs.lognplay.repository.DynamicViewRepository;
import br.pucrs.lognplay.repository.LogSourceRepository;
import br.pucrs.lognplay.repository.RepositoryManager;

public class StorerJob implements Job {

	private static final Logger logger = Logger.getLogger(StorerJob.class);

	private DynamicView dynamicView = null;

	private List<LogSource> logSources = null;

	private ElasticSearchClient esClient;

	public static final String DYNAMIC_VIEW_ID_PARAMETER = "DYNAMIC_VIEW_ID";

	private static final String ES_DATETIME_FORMAT = "YYYY-MM-dd'T'HH:mm:ss.SSS";

	private static final String HDFS_ROOT_FOLDER = "/usr/lognplay/";

	public void initialize(JobExecutionContext jobExecutionContext) {
		logger.trace("initialize job execution...");

		String dynamicViewId = jobExecutionContext.getJobDetail().getJobDataMap().getString(DYNAMIC_VIEW_ID_PARAMETER);
		if (!dynamicViewId.isEmpty()) {
			logger.trace("loading dynamic view");
			try {
				RepositoryManager.beginTransaction();

				DynamicViewRepository dynamicViewRepository = RepositoryManager.getDynamicViewRepository();
				setDynamicView(dynamicViewRepository.get(dynamicViewId));

				LogSourceRepository logSourceRepository = RepositoryManager.getLogSourceRepository();
				setLogSources(logSourceRepository.listByClientId(dynamicView.getClient().getId()));

				RepositoryManager.commitTransaction();
			} catch (Exception e) {
				logger.error(e);
			}
		}

	}

	public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
		try {
			initialize(jobExecutionContext);
			esClient = new ElasticSearchClient();
			esClient.startClient();
			for (LogSource logSource : getLogSources()) {
				processLogSource(logSource);
				removeOldData(logSource);
			}
			esClient.stopClient();
		} catch (Exception e) {
			logger.error(e);
		}
	}

	private void removeOldData(LogSource logSource) {
		logger.trace("removeOldData");
		try {
			Client client = logSource.getClient();
			if (client.getRetentionPeriod() > 0) {
				String clientId = getDynamicView().getClient().getId().toLowerCase().replaceAll("-", "");
				String logSourceId = logSource.getId().toLowerCase().replaceAll("-", "");

				String clientPath = HDFS_ROOT_FOLDER + clientId + Path.SEPARATOR_CHAR;
				String sourcePath = clientPath + logSourceId + Path.SEPARATOR_CHAR;

				Calendar fromDate = Calendar.getInstance();
				fromDate.set(Calendar.HOUR, 0);
				fromDate.set(Calendar.MINUTE, 0);
				fromDate.set(Calendar.SECOND, 0);
				fromDate.set(Calendar.MILLISECOND, 0);
				fromDate.add(Calendar.DAY_OF_MONTH, client.getRetentionPeriod());

				Calendar toDate = (Calendar) fromDate.clone();
				toDate.add(Calendar.HOUR, 1);

				String fileName;
				for (int hour = 0; hour < 24; hour++) {
					fileName = sourcePath + formatFileName(fromDate, toDate);
					HDFSTools.rm(fileName);
					fromDate.add(Calendar.HOUR, 1);
					toDate.add(Calendar.HOUR, 1);
				}

			}
		} catch (Exception e) {
			logger.error(e);
		}
	}

	private void processLogSource(LogSource logSource) {
		try {
			String dynamicViewId = getDynamicView().getId().toLowerCase().replaceAll("-", "");
			String clientId = getDynamicView().getClient().getId().toLowerCase().replaceAll("-", "");
			String logSourceId = logSource.getId().toLowerCase().replaceAll("-", "");

			String clientPath = HDFS_ROOT_FOLDER + clientId + Path.SEPARATOR_CHAR;
			String sourcePath = clientPath + logSourceId + Path.SEPARATOR_CHAR;

			Calendar fromDate = Calendar.getInstance();
			fromDate.set(Calendar.HOUR, 0);
			fromDate.set(Calendar.MINUTE, 0);
			fromDate.set(Calendar.SECOND, 0);
			fromDate.set(Calendar.MILLISECOND, 0);
			fromDate.add(Calendar.DAY_OF_MONTH, -1); // TODO

			HDFSTools.mkdir(clientPath);
			HDFSTools.mkdir(sourcePath);

			for (int hour = 0; hour < 24; hour++) {
				processHour(logSourceId, dynamicViewId, fromDate, sourcePath);
				fromDate.add(Calendar.HOUR, 1);
			}

		} catch (Exception e) {
			logger.trace(e);
		}
	}

	private void processHour(String logSourceId, String dynamicViewId, Calendar fromDate, String sourcePath) {
		Calendar toDate = (Calendar) fromDate.clone();
		toDate.add(Calendar.HOUR, 1);

		String fromDateTime = formatDate(fromDate, ES_DATETIME_FORMAT);
		String toDateTime = formatDate(toDate, ES_DATETIME_FORMAT);

		QueryBuilder query = filteredQuery(termQuery("appName", logSourceId), rangeFilter("timeReported").gt(fromDateTime).lte(toDateTime));
		SearchRequestBuilder builder = esClient.getEsClient().prepareSearch(dynamicViewId);

		builder.addField("_source");
		builder.setSearchType(SearchType.SCAN);
		builder.setQuery(query);
		builder.setFrom(0);
		builder.setSize(1000);
		builder.setScroll(new TimeValue(60000));

		SearchResponse scrollResp = builder.execute().actionGet();

		try {
			FSDataOutputStream fileOut = null;
			String fileName = sourcePath + formatFileName(fromDate, toDate);

			while (true) {
				scrollResp = esClient.getEsClient().prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(600000)).execute().actionGet();

				if (scrollResp.hits().hits().length == 0) {
					break;
				}

				if (fileOut == null) {
					if (HDFSTools.getFileSystem().exists(new Path(fileName))) {
						HDFSTools.rm(fileName);
					}
					fileOut = HDFSTools.getFileSystem().create(new Path(fileName));
				}

				for (SearchHit hit : scrollResp.getHits()) {
					fileOut.writeChars(hit.sourceAsString());
					fileOut.write("\r\n".getBytes());
				}
			}
			if (fileOut != null) {
				fileOut.close();
			}
		} catch (Exception e) {
			logger.error(e);
			e.printStackTrace();
		}
	}

	public DynamicView getDynamicView() {
		return dynamicView;
	}

	public void setDynamicView(DynamicView dynamicView) {
		this.dynamicView = dynamicView;
	}

	public List<LogSource> getLogSources() {
		return logSources;
	}

	public void setLogSources(List<LogSource> logSources) {
		this.logSources = logSources;
	}

	private String formatDate(Calendar date, String pattern) {
		return DateTimeFormat.forPattern(pattern).print(new DateTime(date));
	}

	private String formatFileName(Calendar fromDate, Calendar toDate) {
		String fileName = formatDate(fromDate, "YYYY-MM-dd_HHmm-");
		fileName += formatDate(toDate, "HHmm");
		fileName += ".log";
		return fileName;
	}

}
